package mr.fileToDb;

import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.Map;

import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.OutputCommitter;
import org.apache.hadoop.mapreduce.OutputFormat;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.db.DBConfiguration;
import org.apache.hadoop.mapreduce.lib.db.DBWritable;
import org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import utils.DbConn;
import utils.HdfsUtils;
import utils.IReadFile;

@InterfaceAudience.Public
@InterfaceStability.Stable
public class OracleDBOutputFormat<K extends CollectorWritable, V> extends OutputFormat<K, V> {
	Map<String, String[]> oracleConfigMap = new HashMap<String, String[]>();

	public void checkOutputSpecs(JobContext context) throws IOException, InterruptedException {
	}

	public OutputCommitter getOutputCommitter(TaskAttemptContext context) throws IOException, InterruptedException {
		return new FileOutputCommitter(FileOutputFormat.getOutputPath(context), context);
	}

	@InterfaceStability.Evolving
	public class DBRecordWriter extends RecordWriter<K, V> {
		private Connection connection;
		private Map<String, PreparedStatement> statementMap;

		public DBRecordWriter() throws SQLException {
		}

		public DBRecordWriter(Connection connection, Map<String, PreparedStatement> statementMap) throws SQLException {
			this.connection = connection;
			this.statementMap = statementMap;
		}

		public Connection getConnection() {
			return connection;
		}

		public Map<String, PreparedStatement> getStatementMap() {
			return statementMap;
		}

		public void close(TaskAttemptContext context) throws IOException {
			try {
				for (PreparedStatement statement : statementMap.values()) {
					statement.executeBatch();
				}
			} catch (SQLException e) {
				e.printStackTrace();
				throw new IOException(e.getMessage());
			} finally {
				try {
					for (PreparedStatement statement : statementMap.values()) {
						statement.close();
					}
					connection.close();
					oracleConfigMap.clear();
				} catch (SQLException ex) {
					throw new IOException(ex.getMessage());
				}
			}
		}

		int i = 0;

		public void write(K key, V value) throws IOException {
			PreparedStatement statement = null;
			try {
				i++;
				DBWritable dBWritable = (DBWritable) key.get();
				String className = key.get().getClass().getName().trim();
				String classes[] = className.split("\\.");
				statement = statementMap.get(classes[classes.length - 1]);
				if (statement != null) {
					dBWritable.write(statement);
					statement.addBatch();
					if (i == 500) {
						statement.executeBatch();
						i = 0;
					}
				}
			} catch (SQLException e) {
				e.printStackTrace();
			} finally {
				try {
					statement.executeBatch();
				} catch (SQLException e) {
					e.printStackTrace();
				}
			}
		}
	}

	public String constructQuery(String table, String[] fieldNames) {
		if (fieldNames == null) {
			throw new IllegalArgumentException("Field names may not be null");
		}
		StringBuilder query = new StringBuilder();
		query.append("INSERT INTO ").append(table);
		if (fieldNames.length > 0 && fieldNames[2] != null) {
			query.append("(");
			for (int i = 1; i < fieldNames.length; i++) {

				query.append(fieldNames[i]);
				if (i != fieldNames.length - 1) {
					query.append(",");
				}
			}
			query.append(")");
		}
		query.append(" VALUES(");
		for (int i = 1; i < fieldNames.length; i++) {
			query.append("?");
			if (i != fieldNames.length - 1) {
				query.append(",");
			}
		}
		query.append(")");

		return query.toString();
	}

	public RecordWriter<K, V> getRecordWriter(TaskAttemptContext context) throws IOException {
		FileSystem fs = null;
		String confpath = context.getConfiguration().get("confpath");
		String tableconf = context.getConfiguration().get("tableconf");
		try {
			fs = FileSystem.get(context.getConfiguration());
			HdfsUtils.readPath(fs, confpath, tableconf, new IReadFile() {
				public void paserLine(String[] datas) {
					oracleConfigMap.put(datas[0], datas);
				}
			});
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			fs.close();
		}
		try {
			Map<String, String> map = HdfsUtils.readPath(fs, confpath, "oracle.conf");
			DbConn dbConn = new DbConn(map);
			Connection connection = dbConn.getConnection();
			PreparedStatement statement = null;
			Map<String, PreparedStatement> statementMap = new HashMap<String, PreparedStatement>();
			for (String key : oracleConfigMap.keySet()) {
				String[] tableNameAndFiled = oracleConfigMap.get(key);
				String[] arrs= key.split(":");
				statement = connection.prepareStatement(constructQuery(arrs[1], tableNameAndFiled));
				statementMap.put(arrs[0], statement);
			}
			return new DBRecordWriter(connection, statementMap);
		} catch (Exception ex) {
			ex.printStackTrace();
			throw new IOException(ex.getMessage());
		}
	}

	public static void setOutput(Job job, String tableName, String... fieldNames) throws IOException {
		if (fieldNames.length > 0 && fieldNames[0] != null) {
			DBConfiguration dbConf = setOutput(job, tableName);
			dbConf.setOutputFieldNames(fieldNames);
		} else {
			if (fieldNames.length > 0) {
				setOutput(job, tableName, fieldNames.length);
			} else {
				throw new IllegalArgumentException("Field names must be greater than 0");
			}
		}
	}

	public static void setOutput(Job job, String tableName, int fieldCount) throws IOException {
		DBConfiguration dbConf = setOutput(job, tableName);
		dbConf.setOutputFieldCount(fieldCount);
	}

	private static DBConfiguration setOutput(Job job, String tableName) throws IOException {
		job.setOutputFormatClass(OracleDBOutputFormat.class);
		job.setReduceSpeculativeExecution(false);
		DBConfiguration dbConf = new DBConfiguration(job.getConfiguration());
		dbConf.setOutputTableName(tableName);
		return dbConf;
	}
}
